package com.lefu.risk.storm.topology;


import com.lefu.risk.storm.bolt.CodeCountBolt;
import com.lefu.risk.storm.bolt.RuleMetchBolt;
import com.lefu.risk.storm.spout.LocalRuleSpout;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.kafka.*;
import org.apache.storm.spout.SchemeAsMultiScheme;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.Utils;

import java.util.UUID;


public class RuleTopologyMain {

    public static void main(String[] args) {
        try {
            TopologyBuilder builder = new TopologyBuilder();
            // zk
            BrokerHosts zks = new ZkHosts(
                            "colony1:2181,colony2:2281,colony3:2381," +
                            "colony4:2181,colony5:2281,colony6:2381,colony7:2381");

            SpoutConfig spoutConfig = new SpoutConfig(zks, "riskTest01", "/riskTest01", UUID.randomUUID().toString());
            spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
            KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);

            //集群上用的
//            builder.setSpout("ruleKafkaSpout",kafkaSpout);
//            builder.setBolt("ruleMetchBolt",new RuleMetchBolt(),4).setNumTasks(2).shuffleGrouping("ruleKafkaSpout");

            //本地测试用的
            builder.setSpout("localRuleSpout",new LocalRuleSpout());
            builder.setBolt("ruleMetchBolt",new RuleMetchBolt(),4).setNumTasks(2).shuffleGrouping("localRuleSpout");


            builder.setBolt("codeCountBolt",new CodeCountBolt(),4).setNumTasks(2).shuffleGrouping("ruleMetchBolt");
            Config topologyConf = new Config();
            topologyConf.setDebug(true);
            topologyConf.setMaxSpoutPending(100000);

            if (args != null && args.length > 0) {
                topologyConf.setNumWorkers(5);
                StormSubmitter.submitTopologyWithProgressBar(args[0], topologyConf, builder.createTopology());
            } else {
                topologyConf.setMaxTaskParallelism(4);
                LocalCluster cluster = new LocalCluster();
                cluster.submitTopology("riskTopology", topologyConf, builder.createTopology());
                Utils.sleep(10000000);
                cluster.shutdown();
            }


        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
